package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

public class LinearRegressionReducerWithoutRunningWindow extends Reducer<CodeDateKey, TimestampPriceDataPoint, Text, Text> {


	@Override
	public void reduce(CodeDateKey key, Iterable<TimestampPriceDataPoint> values,
			Context context) throws IOException, InterruptedException {

		Text outKey = new Text();
		Text outVal = new Text();
	
				double sumX = 0;
				double sumY = 0;
				double sumX2 = 0;
				double sumXY = 0;
				int x = 0;
				
				TimestampPriceDataPoint lastDataPoint = null;
				for(TimestampPriceDataPoint dataPoint: values)
				{
					x++;
					double y = dataPoint.getClosePrice();
					sumX += x;
					sumY += y;
					sumX2 += x*x;
					sumXY += x*y;
					lastDataPoint = dataPoint;
				}
				//Slope(b) = (NΣXY - (ΣX)(ΣY)) / (NΣX2 - (ΣX)2)
				 //Intercept(a) = (ΣY - b(ΣX)) / N 
				int n = x;
				double slope = (n*sumXY - sumX*sumY)/(n*sumX2 - sumX*sumX);
				if (Double.isNaN(slope)) slope = 0;
				double intercept =  (sumY -slope*sumX)/n;
				if (Double.isNaN(intercept)) intercept = 0;
				String equation = intercept + " + " + slope + "x";
				
				/*
				outKey.set("Code: " + key.getCode() + ", Date: "
						+ lastDataPoint.getDate());
				outVal.set("Equation: " + equation);
				 */
				outKey.set(key.getCode() + "\t" +lastDataPoint.getDate());
				outVal.set(slope + "\t" + intercept);
				
				context.write(outKey, outVal);

	} // end reduce

}
